package com.lmind.core.gps.gateway;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lmind.core.async.ResultHandler;
import com.lmind.core.gps.terminal.Terminal;
import com.lmind.core.gps.terminal.message.AbstractMessage;
import com.lmind.core.gps.terminal.message.AccessMessage;
import com.lmind.core.gps.terminal.message.KeepAliveMessage;
import com.lmind.core.gps.terminal.message.ReplyMessage;
import com.lmind.core.gps.tm.TerminalSession;
import com.lmind.core.message.AsyncMessageProcessor;
import com.lmind.core.message.MessageService;
import com.lmind.core.message.MessageServiceListener;
import com.lmind.core.message.MessagePipe;
import com.lmind.core.message.PipelineManager;
import com.lmind.core.message.RequestResponseProcessor;

public class SmartTerminalGateway implements MessageServiceListener, RequestResponseProcessor, TerminalGateway {
	
	private static final Logger logger = LoggerFactory.getLogger(SmartTerminalGateway.class);
	
	private Map<String, MessagePipe> pipeBySerial = new ConcurrentHashMap<String, MessagePipe>();
	
	private MessageService messagePipe;
	
	private PipelineManager messageHelper;
	
	private AsyncMessageProcessor<GatewayMessageContext, AbstractMessage> processor;

	public void setMessagePipe(MessageService messagePipe) {
		this.messagePipe = messagePipe;
	}

	@Override
	public void setProcessor(
			AsyncMessageProcessor<GatewayMessageContext, AbstractMessage> processor) {
		this.processor = processor;
	}

	protected void init() {
		
		messagePipe.addMessageServiceListener(this);
		
		messageHelper = new PipelineManager();
		messageHelper.setMessagePipe(messagePipe);
		messageHelper.setMessageProcessor(this);
		messageHelper.init();
	}
	
	@Override
	public void send(String deviceSerial, AbstractMessage message, ResultHandler<ReplyMessage> handler) {
		TerminalSession session = getSession(deviceSerial);
		if (session == null) {
			handler.exception(new Exception("终端:" + deviceSerial + "不在线"));
			return;
		}
		
		class Handler implements ResultHandler<Object> {
			private ResultHandler<ReplyMessage> handler;
			public Handler(ResultHandler<ReplyMessage> handler) {
				this.handler = handler;
			}
			public void result(Object result) {
				handler.result((ReplyMessage)result);
			}
			public void exception(Throwable e) {
				handler.exception(e);
			}
		}
		
		messageHelper.send(session.getMessagePipe(), message, new Handler(handler));
	}

	@Override
	public void connected(MessagePipe pipe) {
		TerminalSession session = new TerminalSession();
		session.setMessagePipe(pipe);
		pipe.setAttribute(TerminalSession.class.getName(), session);
	}

	@Override
	public void disconnected(MessagePipe connection) {
		
		TerminalSession session = getSession(connection);
		if (session == null)
			return;
		
		session.close();
	}

	@Override
	public void message(MessagePipe connection, Object message) {
		
	}

	@Override
	public void keepAlive(MessagePipe connection) {

	}
	

	@Override
	public boolean isRequest(MessagePipe connection, Object message) {
		AbstractMessage msg = (AbstractMessage)message;
		return msg.hasResponse();
	}

	@Override
	public boolean isResponse(MessagePipe connection, Object message) {
		return message instanceof ReplyMessage;
	}
	
	@Override
	public void process(MessagePipe pipe, Object message,
			ResultHandler<Object> handler) {
		
		TerminalSession session = getSession(pipe);
		AbstractMessage msg = (AbstractMessage)message;
		
		// 收到终端的接入消息
		if (message instanceof AccessMessage) {
			AccessMessage access = (AccessMessage)message;
			session.setTerminal(new Terminal(access.getDeviceSerial()));
			pipeBySerial.put(access.getDeviceSerial(), pipe);
			handler.result(new ReplyMessage());
			
			logger.info("终端接入成功:" + access.getDeviceSerial());
		} else if (message instanceof KeepAliveMessage) {
		} else {
			GatewayMessageContext context = new GatewayMessageContext();
			context.setMessagePipe(pipe);
			context.setTerminal(session.getTerminal());
			context.setGateway(this);
			
			class Handler implements ResultHandler<AbstractMessage> {
				private ResultHandler<Object> handler;
				public Handler(ResultHandler<Object> handler) {
					this.handler = handler;
				}
				public void result(AbstractMessage result) {
					handler.result(result);
				}
				public void exception(Throwable e) {
					handler.exception(e);
				}
			}
			
			final ResultHandler<Object> h = handler;
			processor.process(context, (AbstractMessage)message, new Handler(handler));
		}
	}
	
	protected synchronized TerminalSession getSession(String deviceSerial) {
		MessagePipe pipe = pipeBySerial.get(deviceSerial);
		if (pipe == null)
			return null;
		return (TerminalSession)pipe.getAttribute(TerminalSession.class.getName());
	}
	
	protected synchronized TerminalSession getSession(MessagePipe sessionId) {
		return (TerminalSession)sessionId.getAttribute(TerminalSession.class.getName());
	}
	
	/**
	 * 如果存在，关闭会话
	 * @param sessionId
	 */
	protected void closeSession(MessagePipe sessionId) {
		
		TerminalSession session = getSession(sessionId);
		if (session == null)
			return;
		
		session.close();
	}

}
